import asyncio
import logging
from contextlib import contextmanager
from typing import Dict, Union
from redis.asyncio import Redis as AioRedis
from threading import current_thread
import redis

from flask import current_app
from flask_caching import Cache
from flask_redis import FlaskRedis
from flask_restx import abort
from flask_sqlalchemy import SQLAlchemy as BaseSQLAlchemy
from openstack.exceptions import SDKException
from redis import Redis
from werkzeug.exceptions import HTTPException

from configs.settings import flask_config

logger = logging.getLogger(__name__)


class SQLAlchemy(BaseSQLAlchemy):
    @contextmanager
    def auto_commit(self):
        try:
            yield
            self.session.commit()
        except HTTPException as e:  # 主动抛出自定义异常，如abort
            self.session.rollback()
            raise e
        except SDKException as e:
            self.session.rollback()
            current_app.logger.exception("openstack 操作错误", exc_info=e)
            abort(400, message=f"openstack操作异常: {e.message}")
        except Exception as e:
            self.session.rollback()
            current_app.logger.exception(e)
            abort(400, message="未知错误")

    def exists(self, query):
        return self.session.query(query.exists()).scalar()


db = SQLAlchemy()

redis_conn: Union[Redis, FlaskRedis] = FlaskRedis(strict=False)

cache = Cache(
    config={
        "CACHE_TYPE": "redis",
        "CACHE_REDIS_HOST": flask_config.redis_config["host"],
        "CACHE_REDIS_PORT": flask_config.redis_config["port"],
        "CACHE_REDIS_PASSWORD": flask_config.redis_config["password"],
    }
)

class AdpAioRedis:
    redis_pool: Dict[int, AioRedis] = {}

    @property
    def current_thread_id(self) -> int:
        return current_thread().ident

    async def _conn(self, thread_id) -> Union[AioRedis]:
        if self.redis_pool.get(thread_id):
            aio_redis = self.redis_pool[self.current_thread_id]
        else:
            aio_redis = self.pool
            self.redis_pool[thread_id] = aio_redis

        try:
            _ = await aio_redis.ping()
        except RuntimeError:
            loop = asyncio.new_event_loop()
            asyncio.set_event_loop(loop)

            aio_redis = self.pool
            self.redis_pool[thread_id] = aio_redis
            _ = await aio_redis.ping()

        return aio_redis

    @property
    async def conn(self) -> AioRedis:
        thread_id = self.current_thread_id
        async with asyncio.Lock():
            aio_redis = await self._conn(thread_id)
        return aio_redis

    @property
    def pool(self) -> AioRedis:  # noqa
        _pool = AioRedis.from_url(flask_config.REDIS_URL)
        return _pool

    async def execute(self, command, *args, **kw):
        _conn = await self.conn
        return await _conn.execute_command(command, *args, **kw)

    async def publish(self, channel, message, **kwargs):
        _conn = await self.conn
        return await _conn.publish(channel, message, **kwargs)

    async def pubsub(self):
        _conn = await self.conn
        return _conn.pubsub()


aio_redis_pool = AdpAioRedis()

def redis_publist(channel, message, **kwargs):
    host = flask_config.redis_config["host"]
    port = flask_config.redis_config["port"]
    password = flask_config.redis_config["password"]
    
    conn = redis.Redis(host=host, port=port, password=password, db=3)
    print(channel, message)
    conn.publish(channel, message, **kwargs)
